package spring.cloud.tasks.tasks_manager.tasks_manager.service;

import lombok.Getter;
import lombok.Setter;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.StringUtils;
import org.apache.curator.framework.CuratorFramework;
import org.apache.curator.framework.recipes.leader.LeaderLatch;
import org.apache.zookeeper.CreateMode;
import spring.cloud.tasks.tasks_manager.tasks_manager.node.Node;
import spring.cloud.tasks.tasks_manager.tasks_manager.task_item_allot.TasksFullTaskItemTask;
import spring.cloud.tasks.tasks_manager.tasks_manager.task_item_allot.executor.ExecutorLostTrafficTaskItemAllotTask;
import spring.cloud.tasks.tasks_manager.tasks_manager.task_item_allot.executor.ExecutorOfflineTaskItemAllotTask;
import spring.cloud.tasks.tasks_manager.tasks_manager.task_item_allot.executor.ExecutorOnlineTaskItemAllotTask;
import spring.cloud.tasks.tasks_manager.tasks_manager.task_item_allot.executor.ExecutorRecoverTrafficTaskItemAllotTask;
import spring.cloud.tasks.tasks_manager.tasks_manager.task_item_allot.executor.task.TaskExecutorOfflineTaskItemAllotTask;
import spring.cloud.tasks.tasks_manager.tasks_manager.task_item_allot.executor.task.TaskExecutorOnlineTaskItemAllotTask;
import spring.cloud.tasks.tasks_manager.tasks_manager.task_item_allot.task.TaskDisableTaskItemAllotTask;
import spring.cloud.tasks.tasks_manager.tasks_manager.task_item_allot.task.TaskEnableTaskItemAllotTask;
import spring.cloud.tasks.tasks_manager.tasks_manager.task_item_allot.task.TaskExecuteForceTaskItemAllotTask;

import java.nio.charset.StandardCharsets;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.ThreadFactory;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.locks.ReentrantLock;

@Slf4j
@Setter
@Getter
public class AllotTaskItemService {

    public static final boolean CONTAINER_ALIGN_WITH_PHYSICAL;

    private static final String NAME_IS_CONTAINER_ALIGN_WITH_PHYSICAL = "VIP_SATURN_CONTAINER_ALIGN_WITH_PHYSICAL";

    static {
        String isContainerAlignWithPhysicalStr = System.getProperty(NAME_IS_CONTAINER_ALIGN_WITH_PHYSICAL, System.getenv(NAME_IS_CONTAINER_ALIGN_WITH_PHYSICAL));
        CONTAINER_ALIGN_WITH_PHYSICAL = StringUtils.isBlank(isContainerAlignWithPhysicalStr) || Boolean.parseBoolean(isContainerAlignWithPhysicalStr);
    }

    private String namespace;
    private String hostValue;
    private CuratorFramework curatorFramework;
    private AtomicInteger shardingCount;
    private AtomicBoolean needAllSharding;
    private ExecutorService executorService;
    private AllotTaskItemContentService allotTaskItemContentService;
    private ReportAlarmService reportAlarmService;
    private ReentrantLock lock;

    public AllotTaskItemService(CuratorFramework curatorFramework, String hostValue, ReportAlarmService reportAlarmService) {
        this.curatorFramework = curatorFramework;
        this.hostValue = hostValue;
        this.reportAlarmService = reportAlarmService;
        this.shardingCount = new AtomicInteger(0);
        this.needAllSharding = new AtomicBoolean(false);
        this.executorService = newSingleThreadExecutor();
        this.namespace = curatorFramework.getNamespace();
        this.allotTaskItemContentService = new AllotTaskItemContentService(curatorFramework);
        this.lock = new ReentrantLock();
    }

    /**
     * 单线程 SingleThreadExecutor
     *
     * @return
     */
    private ExecutorService newSingleThreadExecutor() {
        return Executors.newSingleThreadExecutor(new ThreadFactory() {
            @Override
            public Thread newThread(Runnable r) {
                return new Thread(r, namespace + "-" + r.getClass().getSimpleName());
            }
        });
    }

    /**
     * 进行全量分片
     */
    public void tasksFullAllot() throws Exception {
        if (isLeadership()) {
            needAllSharding.set(true);
            shardingCount.incrementAndGet();
            executorService.submit(new TasksFullTaskItemTask(this));
            String shardAllAtOnce = Node.getExecutorShardingNodePath("shardAllAtOnce");
            if (curatorFramework.checkExists().forPath(shardAllAtOnce) != null) {
                curatorFramework.delete().deletingChildrenIfNeeded().forPath(shardAllAtOnce);
            }
        }
    }

    /**
     * 作业启用事件
     */
    public void taskEnableAllotTask(String taskId) throws Exception {
        if (isLeadership()) {
            shardingCount.incrementAndGet();
            executorService.submit(new TaskEnableTaskItemAllotTask(this, taskId));
        }
    }

    /**
     * 处理作业禁用事件
     */
    public void taskDisableAllotTask(String taskId) throws Exception {
        if (isLeadership()) {
            shardingCount.incrementAndGet();
            executorService.submit(new TaskDisableTaskItemAllotTask(this, taskId));
        }
    }

    /**
     * 处理作业全排
     */
    public void taskExecuteForceTaskItemAllotTask(String taskId) throws Exception {
        if (isLeadership()) {
            shardingCount.incrementAndGet();
            executorService.submit(new TaskExecuteForceTaskItemAllotTask(this, taskId));
        }
    }

    /**
     * 结点上线处理
     */
    public void executorOnlineAllotTask(String executorId, String ip) throws Exception {
        if (isLeadership()) {
            shardingCount.incrementAndGet();
            executorService.submit(new ExecutorOnlineTaskItemAllotTask(this, executorId, ip));
        }
    }

    /**
     * 结点掉线处理
     */
    public void executorOfflineAllotTask(String executorId) throws Exception {
        if (isLeadership()) {
            shardingCount.incrementAndGet();
            executorService.submit(new ExecutorOfflineTaskItemAllotTask(executorId, this));
        }
    }

    /**
     * 摘取流量
     */
    public void executorLostTrafficAllotTask(String executorId) throws Exception {
        if (isLeadership()) {
            shardingCount.incrementAndGet();
            executorService.submit(new ExecutorLostTrafficTaskItemAllotTask(this, executorId));
        }
    }

    /**
     * 恢复流量
     */
    public void executorRecoverTrafficAllotTask(String executorId) throws Exception {
        if (isLeadership()) {
            shardingCount.incrementAndGet();
            executorService.submit(new ExecutorRecoverTrafficTaskItemAllotTask(this, executorId));
        }
    }


    /**
     * 处理作业executor上线
     */
    public void taskExecutorOnlineAllotTask(String taskId, String executorId) throws Exception {
        if (isLeadership()) {
            shardingCount.incrementAndGet();
            executorService.submit(new TaskExecutorOnlineTaskItemAllotTask(this, executorId, taskId));
        }
    }

    /**
     * 处理作业executor下线
     */
    public void taskExecutorOfflineAllotTask(String taskId, String executorId) throws Exception {
        if (isLeadership()) {
            shardingCount.incrementAndGet();
            executorService.submit(new TaskExecutorOfflineTaskItemAllotTask(this, taskId, executorId));
        }
    }

    /**
     * 判断是否是主节点
     *
     * @return
     * @throws Exception
     */
    private boolean isLeadership() throws Exception {
        while (!hasLeadership()) {
            leaderElection();
        }
        return new String(curatorFramework.getData().forPath(Node.LEADER_HOSTNODE_PATH), StandardCharsets.UTF_8).equals(hostValue);
    }

    /**
     * 选举
     */
    public void leaderElection() throws Exception {
        lock.lockInterruptibly();
        try {
            if (hasLeadership()) {
                return;
            }
            log.info("{}-{} leadership election start", namespace, hostValue);
            try (LeaderLatch leaderLatch = new LeaderLatch(curatorFramework,
                    Node.LEADER_LATCHNODE_PATH)) {
                leaderLatch.start();
                int timeoutSeconds = 60;
                if (leaderLatch.await(timeoutSeconds, TimeUnit.SECONDS)) {
                    if (!hasLeadership()) {
                        becomeLeader();
                    } else {
                        log.info("{}-{} becomes a follower", namespace, hostValue);
                    }
                } else {
                    log.error("{}-{} leadership election is timeout({}s)", namespace, hostValue, timeoutSeconds);
                }
            } catch (InterruptedException e) {
                log.info("{}-{} leadership election is interrupted", namespace, hostValue);
                throw e;
            } catch (Exception e) {
                log.error(namespace + "-" + hostValue + " leadership election error", e);
                throw e;
            }
        } finally {
            lock.unlock();
        }
    }

    private void becomeLeader() throws Exception {
        // 清理、重置变量
        executorService.shutdownNow();
        while (!executorService.isTerminated()) { // 等待全部任务已经退出
            Thread.sleep(100L);
            executorService.shutdownNow();
        }
        needAllSharding.set(false);
        shardingCount.set(0);
        executorService = newSingleThreadExecutor();

        // 持久化$Jobs节点
        if (curatorFramework.checkExists().forPath(Node.TASKS_NODE_PATH) == null) {
            curatorFramework.create().creatingParentsIfNeeded().forPath(Node.TASKS_NODE_PATH);
        }
        // 持久化LeaderValue
        curatorFramework.create().creatingParentsIfNeeded().withMode(CreateMode.EPHEMERAL).forPath(Node.LEADER_HOSTNODE_PATH, hostValue.getBytes(StandardCharsets.UTF_8));

        // 提交全量分片线程
        needAllSharding.set(true);
        shardingCount.incrementAndGet();
        executorService.submit(new TasksFullTaskItemTask(this));
        log.info("{}-{} become leader", namespace, hostValue);
    }

    private boolean hasLeadership() throws Exception {
        return curatorFramework.checkExists().forPath(Node.LEADER_HOSTNODE_PATH) != null;
    }

    public boolean isLeadershipOnly() throws Exception {
        if (hasLeadership()) {
            return new String(curatorFramework.getData().forPath(Node.LEADER_HOSTNODE_PATH), StandardCharsets.UTF_8).equals(hostValue);
        } else {
            return false;
        }
    }

    private void releaseMyLeadership() throws Exception {
        if (isLeadershipOnly()) {
            curatorFramework.delete().forPath(Node.LEADER_HOSTNODE_PATH);
        }
    }


    public void shutdown() throws InterruptedException {
        shutdownInner(true);
    }

    /**
     * Firstly, shutdown thread pool. Secondly, release my leadership.
     *
     * @param shutdownNow true, the thread pool will be shutdownNow; false, it will be shutdown
     */
    public void shutdownInner(boolean shutdownNow) throws InterruptedException {
        lock.lockInterruptibly();
        try {
            if (executorService != null) {
                if (shutdownNow) {
                    executorService.shutdownNow();
                } else {
                    executorService.shutdown();
                }
            }
            try {
                if (curatorFramework.getZookeeperClient().isConnected()) {
                    releaseMyLeadership();
                }
            } catch (Exception e) {
                log.error(namespace + "-" + hostValue + " delete leadership error", e);
            }
        } finally {
            lock.unlock();
        }
    }

    public int shardingCountIncrementAndGet() {
        return shardingCount.incrementAndGet();
    }

    public int shardingCountDecrementAndGet() {
        return shardingCount.decrementAndGet();
    }

    public void setNeedAllSharding(boolean needAllSharding) {
        this.needAllSharding.set(needAllSharding);
    }

    public boolean isNeedAllSharding() {
        return needAllSharding.get();
    }

}
